//
// Created by 王一飞 on 2021/4/10.
//
#include <stdbool.h>
#include <amqp.h>
#include <amqp_tcp_socket.h>
#include "head.h"
#include "thread_pool.h"
#include "rabbitmq_api.h"
#include "rabbitmq_utils.h"

const char queues[6][128] = {CPUQUEUE,
                       DISKQUEUE,
                       MEMQUEUE,
                       PROCQUEUE,
                       SYSQUEUE,
                       USERQUEUE};

threadpool_server *threadpool_create(int min_thr_num, int max_thr_num) {
    int i;
    threadpool_server *pool = NULL;
    do {
        /* 线程池空间开辟 */
        if ((pool = (threadpool_server *) malloc(sizeof(threadpool_server))) == NULL) {
            printf("malloc threadpool false; \n");
            break;
        }
        /*信息初始化*/
        pool->min_thr_num = min_thr_num;
        pool->max_thr_num = max_thr_num;
        pool->busy_thr_num = 0;
        pool->live_thr_num = min_thr_num;
        pool->wait_exit_thr_num = 0;
        pool->shutdown = false;
        pool->messCnt = 0;
        pool->threads = (pthread_t *)malloc(sizeof(pthread_t) * pool->max_thr_num);
        memset(pool->threads, 0, sizeof(pthread_t) * max_thr_num);

        /* 初始化互斥锁和条件变量 */
        // if (pthread_mutex_init(&(pool->lock), NULL) != 0 ||
        //     pthread_mutex_init(&(pool->thread_counter), NULL) != 0 ||
        //     pthread_cond_init(&(pool->queue_not_empty), NULL) != 0 ||
        //     pthread_cond_init(&(pool->queue_not_full), NULL) != 0) {
        //     DBG(ERROR "init lock or cond false;");
        //     // printf("init lock or cond false;\n");
        //     break;
        // }
        // 互斥锁的初始化
        pthread_mutexattr_t attr;
        pthread_mutexattr_init(&attr);
        pthread_mutexattr_setpshared(&attr, PTHREAD_PROCESS_SHARED);
        pthread_mutex_init(&(pool->lock), &attr);

        // 条件变量的初始化
        pthread_condattr_t c_attr;
        pthread_condattr_init(&c_attr);
        pthread_condattr_setpshared(&c_attr, PTHREAD_PROCESS_SHARED);
        pthread_cond_init(&(pool->queue_not_empty), &c_attr);

        // rabbitmq确认message个数线程
        DBG(INFO"before cnt thread");
        pthread_create(&(pool->message_tid), NULL, rabbitmq_thread, (void *) pool);
        // sleep(2);

        DBG(INFO"before work thread");
        /* 启动min_thr_num个工作线程 */
        for (i = 0; i < min_thr_num; i++) {
            /* pool指向当前线程池  threadpool_thread函数在后面讲解 */
            pthread_create(&(pool->threads[i]), NULL, threadpool_thread, (void *) pool);
            DBG(INFO "start thread 0x%x... ", (unsigned int) pool->threads[i]);
        }

        // DBG(INFO"before admin thread");
        // /* 管理者线程 admin_thread函数在后面讲解 */
        // pthread_create(&(pool->admin_tid), NULL, admin_thread, (void *) pool);

        return pool;
    } while (0);

    /* 释放pool的空间 */
    threadpool_free(pool);
    return NULL;
}

/*释放线程池*/
int threadpool_free(threadpool_server *pool) {
    if (pool == NULL)
        return -1;
    if (pool->threads) {
        free(pool->threads);
        pthread_mutex_lock(&(pool->lock)); /*先锁住再销毁*/
        pthread_mutex_destroy(&(pool->lock));
        pthread_mutex_lock(&(pool->thread_counter));
        pthread_mutex_destroy(&(pool->thread_counter));
        pthread_cond_destroy(&(pool->queue_not_empty));
        pthread_cond_destroy(&(pool->queue_not_full));
    }
    free(pool);
    pool = NULL;

    return 0;
}

int getSelfChannel(threadpool_server *threadpool) {
    pthread_t tid = pthread_self();
    int channel = 0;
    for (int i = 0; i < threadpool->max_thr_num; i++) {
        if (threadpool->threads[i] == tid) {
            // channel = threadpool->channels[i];
            channel = i + 1;
            break;
        }
    }
    return channel;
}

// void *threadpool_thread(void *threadpool) {
//     threadpool_server *pool = (threadpool_server *) threadpool;
//     amqp_connection_state_t conn = connectServer("47.111.117.240", 5672);
//     int channel = 1;
//     amqp_channel_open(conn, channel);
//     die_on_amqp_error(amqp_get_rpc_reply(conn), "Opening channel");
//
//
//     while (1) {
//         // DBG(INFO"<0x%x> before lock" , (unsigned int)pthread_self());
//         pthread_mutex_lock(&(pool->lock));
//         DBG(INFO"<0x%x> get lock", (unsigned int)pthread_self());
//         /* 无任务则阻塞在 “任务队列不为空” 上，有任务则跳出 */
//         while ((pool->messCnt == 0) && (!pool->shutdown)) {
//             DBG(INFO"thread 0x%x is waiting ", (unsigned int)pthread_self())
//             pthread_cond_wait(&(pool->queue_not_empty), &(pool->lock));
//
//             /* 判断是否需要清除线程,自杀功能 */
//             if (pool->wait_exit_thr_num > 0) {
//                 pool->wait_exit_thr_num--;
//                 /* 判断线程池中的线程数是否大于最小线程数，是则结束当前线程 */
//                 if (pool->live_thr_num > pool->min_thr_num) {
//                     DBG(INFO"thread 0x%x is exiting ", (unsigned int) pthread_self())
//                     pool->live_thr_num--;
//                     pthread_mutex_unlock(&(pool->lock));
//                     pthread_exit(NULL); //结束线程
//                 }
//             }
//         }
//         /* 线程池开关状态 */
//         //关闭线程池
//         if (pool->shutdown) {
//             pthread_mutex_unlock(&(pool->lock));
//             DBG(INFO"thread 0x%x is exiting ", (unsigned int) pthread_self())
//             pthread_exit(NULL); //线程自己结束自己
//         }
//         // 取出消息
//         amqp_basic_qos(conn, channel, 0, 20, 0);
//         amqp_basic_consume(conn, channel, amqp_cstring_bytes("slave-health-mem"), amqp_empty_bytes,
//                            0, 0, 0, amqp_empty_table);
//         die_on_amqp_error(amqp_get_rpc_reply(conn), "Consuming");
//
//
//         pool->busy_thr_num++;
//         struct timeval timeInter;
//         timeInter.tv_sec = 0;
//         timeInter.tv_usec = 500000;
//         amqp_envelope_t envelope;
//         amqp_rpc_reply_t res;
//         // 从6个queue中一次尝试取message
//         for (int i = 2; i < 3; i++) {
//             amqp_maybe_release_buffers(conn);
//             res = amqp_consume_message(conn, &envelope, NULL, 0);
//             if (AMQP_RESPONSE_NORMAL == res.reply_type) {
//                 // 如果取到了消息就break;
//                 pool->messCnt--;
//                 break;
//             }
//         }
//         pool->busy_thr_num--;
//         //释放线程锁
//         pthread_mutex_unlock(&(pool->lock));
//         // 如果从6个队列中都没有取出消息则continue,不执行以下的代码
//         if (AMQP_RESPONSE_NORMAL != res.reply_type) {
//             amqp_destroy_envelope(&envelope);
//             printf("error in res\n");
//             continue;
//         }
//         // 从信封中取出消息
//         char *message = (char *)calloc(envelope.message.body.len + 1, sizeof(char));
//         strncpy(message, envelope.message.body.bytes, envelope.message.body.len);
//         printf(L_PINK"<0x%x> message: %s\n"NONE, (unsigned int)pthread_self(), message);
//
//         // TODO: 执行mysql的操作
//         // pool->busy_thr_num--;
//         // 如果插入mysql成功的话,还需要应答消息
//         // DBG(INFO"before ack");
//         // amqp_basic_ack(pool->conn, channel, envelope.delivery_tag, 0);
//         // sleep(1);
//         // amqp_dump(envelope.message.body.bytes, envelope.message.body.len);
//         if (amqp_basic_ack(conn, channel, envelope.delivery_tag, 0) == 0) {
//             printf("success on ack\n");
//         } else {
//             printf("error on ack\n");
//         }
//         amqp_destroy_envelope(&envelope);
//         free(message);
//     }
//     pthread_exit(NULL);
// }

void *admin_thread(void *threadpool) {

}

_Noreturn void *rabbitmq_thread(void *threadpool) {
    amqp_connection_state_t conn = connectServer("47.111.117.240", 5672);
    amqp_channel_open(conn, 1000);
    threadpool_server *pool = (threadpool_server *) threadpool;
    while (1) {
        pthread_mutex_lock(&(pool->lock));
        uint32_t messCnt = 0;
        // char queues[6][128] = {CPUQUEUE,
        //                        DISKQUEUE,
        //                        MEMQUEUE,
        //                        PROCQUEUE,
        //                        SYSQUEUE,
        //                        USERQUEUE};
        DBG(INFO"before loop");
        for (int i = 2; i < 3; i++) {
            // amqp_queue_declare_ok_t *r = amqp_queue_declare(pool->conn, pool->max_thr_num + 1, amqp_cstring_bytes(queues[i]), 0, 1, 0, 0, amqp_empty_table);
            amqp_queue_declare_ok_t *r = amqp_queue_declare(conn, 1000, amqp_cstring_bytes(queues[i]), 0, 1, 0, 0, amqp_empty_table);
            die_on_amqp_error(amqp_get_rpc_reply(conn), "Declaring queue");
            messCnt += r->message_count;

        }
        // amqp_channel_close(conn, 1, AMQP_REPLY_SUCCESS);
        // amqp_connection_close(conn, AMQP_REPLY_SUCCESS);
        // amqp_destroy_connection(conn);

        pool->messCnt = messCnt;
        if (messCnt > 0)
            pthread_cond_broadcast(&(pool->queue_not_empty));
        pthread_mutex_unlock(&(pool->lock));
        printf(INFO"message count: %d\n", messCnt);
        DBG(INFO"after get message count");

        sleep(10);
    }
    pthread_exit(NULL);
}


void *threadpool_thread(void *threadpool) {
    threadpool_server *pool = (threadpool_server *)threadpool;
    // threadpool_server *pool = (threadpool_server *) threadpool;
    char hostname[128] = "localhost";
    int port = 5672, status;
    amqp_socket_t *socket = NULL;
    amqp_connection_state_t conn;
    char queuename[128] = "slave-health-mem";

    conn = amqp_new_connection();

    socket = amqp_tcp_socket_new(conn);
    if (!socket) {
        die("creating TCP socket");
    }

    status = amqp_socket_open(socket, hostname, port);
    if (status) {
        die("opening TCP socket");
    }

    die_on_amqp_error(amqp_login(conn, "/", 0, 131072, 0, AMQP_SASL_METHOD_PLAIN,
                                 "admin", "admin"),
                      "Logging in");
    amqp_channel_open(conn, 1);
    die_on_amqp_error(amqp_get_rpc_reply(conn), "Opening channel");

    while (1) {
        // DBG(INFO"<0x%x> before lock" , (unsigned int)pthread_self());
        pthread_mutex_lock(&(pool->lock));
        // DBG(INFO"<0x%x> get lock", (unsigned int)pthread_self());
        /* 无任务则阻塞在 “任务队列不为空” 上，有任务则跳出 */
        while ((pool->messCnt == 0) && (!pool->shutdown)) {
            DBG(INFO"thread 0x%x is waiting ", (unsigned int)pthread_self())
            DBG(INFO"<0x%x> get lock", (unsigned int)pthread_self());
            pthread_cond_wait(&(pool->queue_not_empty), &(pool->lock));
            /* 判断是否需要清除线程,自杀功能 */
            // if (pool->wait_exit_thr_num > 0) {
            //     pool->wait_exit_thr_num--;
            //     /* 判断线程池中的线程数是否大于最小线程数，是则结束当前线程 */
            //     if (pool->live_thr_num > pool->min_thr_num) {
            //         DBG(INFO"thread 0x%x is exiting ", (unsigned int) pthread_self())
            //         pool->live_thr_num--;
            //         pthread_mutex_unlock(&(pool->lock));
            //         pthread_exit(NULL); //结束线程
            //     }
            // }
        }
        /* 线程池开关状态 */
        //关闭线程池
        // if (pool->shutdown) {
        //     pthread_mutex_unlock(&(pool->lock));
        //     DBG(INFO"thread 0x%x is exiting ", (unsigned int) pthread_self())
        //     pthread_exit(NULL); //线程自己结束自己
        // }
        // 取出消息
        amqp_basic_consume(conn, 1, amqp_cstring_bytes(queuename), amqp_empty_bytes,
                           0, 0, 0, amqp_empty_table);
        die_on_amqp_error(amqp_get_rpc_reply(conn), "Consuming");
        amqp_rpc_reply_t res;
        amqp_envelope_t envelope;

        amqp_maybe_release_buffers(conn);

        struct timeval timeInter;
        timeInter.tv_sec = 0;
        timeInter.tv_usec = 0;

        res = amqp_consume_message(conn, &envelope, NULL, 0);

        if (AMQP_RESPONSE_NORMAL != res.reply_type) {
            break;
        }
        pool->messCnt--;
        pthread_mutex_unlock(&(pool->lock));
        printf("Delivery %u, exchange %.*s routingkey %.*s\n",
               (unsigned)envelope.delivery_tag, (int)envelope.exchange.len,
               (char *)envelope.exchange.bytes, (int)envelope.routing_key.len,
               (char *)envelope.routing_key.bytes);

        if (envelope.message.properties._flags & AMQP_BASIC_CONTENT_TYPE_FLAG) {
            printf("Content-type: %.*s\n",
                   (int)envelope.message.properties.content_type.len,
                   (char *)envelope.message.properties.content_type.bytes);
        }
        printf("----\n");
        // pthread_mutex_lock(&(pool->lock));
        pool->messCnt--;
        // pthread_mutex_unlock(&(pool->lock));
        amqp_dump(envelope.message.body.bytes, envelope.message.body.len);
        amqp_basic_ack(conn, 1, envelope.delivery_tag, 0);
        amqp_destroy_envelope(&envelope);



        // pool->busy_thr_num--;
        //释放线程锁
        // pthread_mutex_unlock(&(pool->lock));
        // 如果从6个队列中都没有取出消息则continue,不执行以下的代码
    }

}
